package org.snail.transport.impl;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.HashedWheelTimer;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import org.snail.SnailClient;
import org.snail.common.SnailThreadFactory;
import org.snail.transport.Connection;
import org.snail.transport.Connector;
import org.snail.transport.ConnectorDispatcher;
import org.snail.transport.handler.*;

import java.util.concurrent.ThreadFactory;

/**
 * Created by fish on 17/8/20.
 */
public class NettyConnector implements Connector{

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NettyConnector.class);

    protected final HashedWheelTimer timer = new HashedWheelTimer(new SnailThreadFactory("connector.timer"));

    protected final SnailEncoder encoder = new SnailEncoder();

    protected final ConnectorIdleStateTrigger idleStateTrigger = new ConnectorIdleStateTrigger();

    protected final ConnectionWatcher watcher = new ConnectionWatcher(this);

    protected final ConnectorHandler handler = new ConnectorHandler(this);

    private String host;

    private int port;

    @Override
    public void connect(String host, int port) {
        this.host = host;
        this.port = port;
        int workerNum = Runtime.getRuntime().availableProcessors();

        ThreadFactory workerFactory = new SnailThreadFactory("snail.connector.worker", Thread.MAX_PRIORITY);
        EventLoopGroup worker = new NioEventLoopGroup(workerNum, workerFactory);

        final Bootstrap bootstrap = new Bootstrap().group(worker);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
        bootstrap.option(ChannelOption.SO_REUSEADDR, true);
        bootstrap.option(ChannelOption.SO_REUSEADDR, true);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        bootstrap.option(ChannelOption.ALLOW_HALF_CLOSURE, false);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new IdleStateHandler(timer, 0, 30, 0));//心跳检测
                ch.pipeline().addLast(idleStateTrigger);                     //心跳触发
                ch.pipeline().addLast(watcher);                              //重连监听
                ch.pipeline().addLast(new SnailDecoder());                   //解码
                ch.pipeline().addLast(encoder);                              //编码
                ch.pipeline().addLast(handler);                              //业务处理
            }
        });

        ChannelFuture future =  bootstrap.connect(host,port);

        //连接监听
        future.addListener(new ConnectionListener(this,timer));
        try {
            future.sync();
        } catch (InterruptedException e) {
            logger.error(e);
        }
        Connection connection =  new Connection(future.channel());
        SnailClient.getInstance().addConnection(host,port,connection);
    }


    @Override
    public void bindDispatcher(ConnectorDispatcher dispatcher) {
        this.handler.setDispatcher(dispatcher);
    }

    public String getHost() {
        return host;
    }

    public int getPort() {
        return port;
    }

}
